package com.quanyan.ic.datacenter.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.ic.common.model.OrderMessage;
import com.quanyan.ic.datacenter.constant.MqConstants;
import com.quanyan.ic.datacenter.dao.PlaceDao;
import com.quanyan.ic.datacenter.dao.PlaceOrderDao;
import com.quanyan.ic.datacenter.model.PlaceUnitInfo;
import com.quanyan.ic.datacenter.mq.service.ConsumerOrderMqService;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.AbstractMqConsumer;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.MqConsumerStatus;
import com.quanyan.ic.middleware.mq.rocketmq.consumer.MqTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.List;

/**
 * MQ consumer，消费：
 * 场馆订单对应的场地智能灯自动控制消息
 */
@Component
public class PlaceUnitOrderConsumer extends AbstractMqConsumer {

    private static Logger logger = LoggerFactory.getLogger(PlaceUnitOrderConsumer.class);

    @Value("${rocketmq.nameServer}")
    private String nameServer;

    @Autowired
    private ConsumerOrderMqService consumerOrderMqService;

    @Autowired
    private PlaceOrderDao placeOrderDao;

    @Autowired
    private PlaceDao placeDao;


    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @PostConstruct
    public void start(){
//        super.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
//        super.setConsumeTimestamp("20170824080000");
        super.start();
        logger.info("PlaceUnitOrderConsumer started!");
    }

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        logger.info("received PlaceUnitOrder MQ msg: {}", messages);
        if (messages.isEmpty()){
            return MqConsumerStatus.CONSUMER_SUCCESS;
        }
        MessageExt messageExt = messages.get(0);
        String msgId = messageExt.getKeys();        //订单号
        byte[] bodyByte = messageExt.getBody();
        String j_orderMsg = new String(bodyByte);
        logger.info("PlaceUnitOrder: msgId={}, msg={}", msgId, j_orderMsg);
        OrderMessage orderMessage = null;
        PlaceUnitInfo placeUnitInfo = null;
        try {
            orderMessage = JSON.parseObject(j_orderMsg, OrderMessage.class);
            if (orderMessage.getEndTime() <= sdf.parse("2017-08-25 00:00:00").getTime()){
                return MqConsumerStatus.CONSUMER_SUCCESS;
            }
            placeUnitInfo = placeDao.getPlaceUnitInfo(orderMessage.getSiteId());
            placeOrderDao.insert(orderMessage,placeUnitInfo);//记录订单入库
        }catch (Exception e) {
            logger.error("PlaceUnitOrder error:{}",e);
        }
        try {
            consumerOrderMqService.consumerOrderMsg(orderMessage,placeUnitInfo);
        } catch (Exception e) {
            e.printStackTrace();
        }

        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

    @Override
    protected String getConsumerGroup() {
        return MqConstants.GROP_C_PLACE_UNIT_ORDER;
    }

    @Override
    protected String getNameServerAddress() {
        return nameServer;
    }

    @Override
    protected String getInstanceName() {
        return null;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(
                MqConstants.TOPIC_PLACE_UNIT_ORDER, MqConstants.TAG_PLACE_UNIT_ORDER);
        return mqTopics;
    }
}
